package com.sjc.zeye.product.function;

import com.sjc.zeye.product.domain.ProductViewCount;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 指定格式输出 ：将每个 key每个窗口聚合后的结果带上其他信息进行输出  进入的数据为Long 返回 ItemViewCount对象
 */
public class WindowResult implements WindowFunction<Long, ProductViewCount,Tuple, TimeWindow> {
    @Override
    public void apply(Tuple key, // 窗口的主键，即 itemId
                      TimeWindow timeWindow,  // 窗口
                      Iterable<Long> input,  // 聚合函数的结果，即 count 值
                      Collector<ProductViewCount> out) throws Exception { // 输出类型为 ProductViewCount
        long itemId = ((Tuple1<Long>) key).f0;
        out.collect(new ProductViewCount(itemId,timeWindow.getEnd(),input.iterator().next()));
    }
}
